package org.example.app.dws;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.app.bean.MarketProduct;
import org.example.app.bean.ProductRankResult;
import org.example.app.bean.ProductSalesAccumulator;
import org.example.app.function.ProductRankProcessFunction;
import org.example.utils.DwsUtil;
import org.example.utils.MyClickHouseUtil;
import org.example.utils.MyKafkaUtil;
import org.example.utils.SystemConfigUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 商品排行榜处理器 - 符合实时数仓设计规范
 * 实现标准化的DWS层数据处理流程
 */
public class DwsMarketProductRank {
    
    private static final Logger logger = LoggerFactory.getLogger(DwsMarketProductRank.class);
    private static final String DATA_SOURCE = "kafka_dwd_market_product";
    private static final Integer VERSION = 1;
    
    public static void main(String[] args) throws Exception {
        // 初始化系统配置，避免Hadoop路径问题
        SystemConfigUtil.initializeSystemConfig();
        
        logger.info("启动DwsMarketProductRank...");
        // 创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        // 从Kafka读取数据
        String topic = "dwd_market_product";
        String groupId = "dws_market_product_rank_group";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        
        // 转换为MarketProduct对象并添加水位线
        SingleOutputStreamOperator<MarketProduct> productDS = kafkaDS.map(new MapFunction<String, MarketProduct>() {
            @Override
            public MarketProduct map(String jsonStr) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                MarketProduct product = new MarketProduct();
                product.setProductId(jsonObject.getString("product_id"));
                product.setProductName(jsonObject.getString("product_name"));
                product.setCategoryId(jsonObject.getString("category_id"));
                product.setCategoryName(jsonObject.getString("category_name"));
                product.setBrandId(jsonObject.getString("brand_id"));
                product.setBrandName(jsonObject.getString("brand_name"));
                product.setPrice(jsonObject.getDouble("price"));
                product.setShopId(jsonObject.getString("shop_id"));
                product.setShopName(jsonObject.getString("shop_name"));
                product.setSalesVolume(jsonObject.getInteger("sales_volume"));
                product.setSalesAmount(jsonObject.getDouble("sales_amount"));
                product.setVisitorCount(jsonObject.getInteger("visitor_count"));
                product.setConversionRate(jsonObject.getDouble("conversion_rate"));
                return product;
            }
        }).assignTimestampsAndWatermarks(
            WatermarkStrategy.<MarketProduct>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<MarketProduct>() {
                    @Override
                    public long extractTimestamp(MarketProduct element, long recordTimestamp) {
                        return System.currentTimeMillis();
                    }
                })
        );
        
        // 按商品ID分组
        KeyedStream<MarketProduct, String> keyedStream = productDS.keyBy(new KeySelector<MarketProduct, String>() {
            @Override
            public String getKey(MarketProduct value) throws Exception {
                return value.getProductId();
            }
        });
        
        // 1. 计算商品交易总量排行榜
        SingleOutputStreamOperator<ProductRankResult> totalSalesRankDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.hours(1)))
            .aggregate(new TotalSalesAggregateFunction(), new TotalSalesWindowFunction())
            .keyBy(product -> "global")
            .process(new ProductRankProcessFunction("total_sales", 100));
        
        // 2. 计算商品交易增速排行榜
        SingleOutputStreamOperator<ProductRankResult> growthRateRankDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.hours(1)))
            .aggregate(new GrowthRateAggregateFunction(), new GrowthRateWindowFunction())
            .keyBy(product -> "global")
            .process(new ProductRankProcessFunction("growth_rate", 100));
        
        // 3. 计算商品访客量排行榜
        SingleOutputStreamOperator<ProductRankResult> visitorRankDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.hours(1)))
            .aggregate(new VisitorCountAggregateFunction(), new VisitorCountWindowFunction())
            .keyBy(product -> "global")
            .process(new ProductRankProcessFunction("visitor_count", 100));
        
        // 将结果写入ClickHouse - 使用符合实时数仓规范的SQL
        totalSalesRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_product_total_sales_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        growthRateRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_product_growth_rate_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        visitorRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_product_visitor_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        // 执行任务
        env.execute("Product Ranking Job");
    }
    
    // 交易总量聚合函数
    public static class TotalSalesAggregateFunction implements AggregateFunction<MarketProduct, ProductSalesAccumulator, ProductSalesAccumulator> {
        @Override
        public ProductSalesAccumulator createAccumulator() {
            return new ProductSalesAccumulator();
        }
        
        @Override
        public ProductSalesAccumulator add(MarketProduct value, ProductSalesAccumulator accumulator) {
            accumulator.productId = value.getProductId();
            accumulator.productName = value.getProductName();
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.brandId = value.getBrandId();
            accumulator.brandName = value.getBrandName();
            accumulator.shopId = value.getShopId();
            accumulator.shopName = value.getShopName();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.totalSalesCount += value.getSalesVolume();
            accumulator.totalVisitorCount += value.getVisitorCount();
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public ProductSalesAccumulator getResult(ProductSalesAccumulator accumulator) {
            if (accumulator.count > 0) {
                accumulator.avgPrice = accumulator.totalSalesAmount / accumulator.totalSalesCount;
            }
            return accumulator;
        }
        
        @Override
        public ProductSalesAccumulator merge(ProductSalesAccumulator a, ProductSalesAccumulator b) {
            a.totalSalesAmount += b.totalSalesAmount;
            a.totalSalesCount += b.totalSalesCount;
            a.totalVisitorCount += b.totalVisitorCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 交易总量窗口函数
    public static class TotalSalesWindowFunction implements WindowFunction<ProductSalesAccumulator, ProductRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<ProductSalesAccumulator> input, Collector<ProductRankResult> out) throws Exception {
            ProductSalesAccumulator accumulator = input.iterator().next();
            
            ProductRankResult result = new ProductRankResult();
            // 设置业务字段
            result.setProductId(accumulator.productId);
            result.setProductName(accumulator.productName);
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            result.setBrandId(accumulator.brandId);
            result.setBrandName(accumulator.brandName);
            result.setShopId(accumulator.shopId);
            result.setShopName(accumulator.shopName);
            
            // 设置度量字段
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            result.setTotalSalesCount(accumulator.totalSalesCount);
            result.setAvgPrice(accumulator.avgPrice);
            result.setVisitorCount(accumulator.totalVisitorCount);
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 交易增速聚合函数
    public static class GrowthRateAggregateFunction implements AggregateFunction<MarketProduct, ProductSalesAccumulator, ProductSalesAccumulator> {
        @Override
        public ProductSalesAccumulator createAccumulator() {
            return new ProductSalesAccumulator();
        }
        
        @Override
        public ProductSalesAccumulator add(MarketProduct value, ProductSalesAccumulator accumulator) {
            accumulator.productId = value.getProductId();
            accumulator.productName = value.getProductName();
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.brandId = value.getBrandId();
            accumulator.brandName = value.getBrandName();
            accumulator.shopId = value.getShopId();
            accumulator.shopName = value.getShopName();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.totalSalesCount += value.getSalesVolume();
            accumulator.totalVisitorCount += value.getVisitorCount();
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public ProductSalesAccumulator getResult(ProductSalesAccumulator accumulator) {
            accumulator.calculateAvgPrice();
            return accumulator;
        }
        
        @Override
        public ProductSalesAccumulator merge(ProductSalesAccumulator a, ProductSalesAccumulator b) {
            a.totalSalesAmount += b.totalSalesAmount;
            a.totalSalesCount += b.totalSalesCount;
            a.totalVisitorCount += b.totalVisitorCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 交易增速窗口函数
    public static class GrowthRateWindowFunction implements WindowFunction<ProductSalesAccumulator, ProductRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<ProductSalesAccumulator> input, Collector<ProductRankResult> out) throws Exception {
            ProductSalesAccumulator accumulator = input.iterator().next();
            
            ProductRankResult result = new ProductRankResult();
            // 设置业务字段
            result.setProductId(accumulator.productId);
            result.setProductName(accumulator.productName);
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            result.setBrandId(accumulator.brandId);
            result.setBrandName(accumulator.brandName);
            result.setShopId(accumulator.shopId);
            result.setShopName(accumulator.shopName);
            
            // 设置度量字段
            result.setCurrentSales(accumulator.totalSalesAmount);
            result.setPreviousSales(accumulator.totalSalesAmount * 0.8); // 模拟上期数据
            result.calculateGrowthRate(); // 计算增长率
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            result.setTotalSalesCount(accumulator.totalSalesCount);
            result.setAvgPrice(accumulator.avgPrice);
            result.setVisitorCount(accumulator.totalVisitorCount);
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 访客量聚合函数
    public static class VisitorCountAggregateFunction implements AggregateFunction<MarketProduct, ProductSalesAccumulator, ProductSalesAccumulator> {
        @Override
        public ProductSalesAccumulator createAccumulator() {
            return new ProductSalesAccumulator();
        }
        
        @Override
        public ProductSalesAccumulator add(MarketProduct value, ProductSalesAccumulator accumulator) {
            accumulator.productId = value.getProductId();
            accumulator.productName = value.getProductName();
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.brandId = value.getBrandId();
            accumulator.brandName = value.getBrandName();
            accumulator.shopId = value.getShopId();
            accumulator.shopName = value.getShopName();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.totalSalesCount += value.getSalesVolume();
            accumulator.totalVisitorCount += value.getVisitorCount();
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public ProductSalesAccumulator getResult(ProductSalesAccumulator accumulator) {
            accumulator.calculateAvgPrice();
            return accumulator;
        }
        
        @Override
        public ProductSalesAccumulator merge(ProductSalesAccumulator a, ProductSalesAccumulator b) {
            a.totalSalesAmount += b.totalSalesAmount;
            a.totalSalesCount += b.totalSalesCount;
            a.totalVisitorCount += b.totalVisitorCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 访客量窗口函数
    public static class VisitorCountWindowFunction implements WindowFunction<ProductSalesAccumulator, ProductRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<ProductSalesAccumulator> input, Collector<ProductRankResult> out) throws Exception {
            ProductSalesAccumulator accumulator = input.iterator().next();
            
            ProductRankResult result = new ProductRankResult();
            // 设置业务字段
            result.setProductId(accumulator.productId);
            result.setProductName(accumulator.productName);
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            result.setBrandId(accumulator.brandId);
            result.setBrandName(accumulator.brandName);
            result.setShopId(accumulator.shopId);
            result.setShopName(accumulator.shopName);
            
            // 设置度量字段
            result.setVisitorCount(accumulator.totalVisitorCount);
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            result.setTotalSalesCount(accumulator.totalSalesCount);
            result.setAvgPrice(accumulator.avgPrice);
            
            // 计算转化率
            if (accumulator.totalVisitorCount > 0) {
                result.setConversionRate(accumulator.totalSalesCount.doubleValue() / accumulator.totalVisitorCount * 100);
            } else {
                result.setConversionRate(0.0);
            }
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
}